CI MESSAGE: [7183171]: BUILD STARTED

CI MESSAGE: [7183171]: BUILD FAILED

CI MESSAGE: [7189340]: BUILD STARTED

CI MESSAGE: [7189340]: BUILD PASSED

Check out this pull request on 

See visual diffs & provide feedback on Jupyter Notebooks.

Powered by ReviewNB

CI MESSAGE: [41172120]: BUILD STARTED

CI MESSAGE: [41172120]: BUILD FAILED

CI MESSAGE: [41183936]: BUILD STARTED

CI MESSAGE: [41184260]: BUILD STARTED

CI MESSAGE: [41187088]: BUILD STARTED

CI MESSAGE: [41187088]: BUILD FAILED

CI MESSAGE: [41571534]: BUILD STARTED

CI MESSAGE: [41571534]: BUILD FAILED

CI MESSAGE: [41726080]: BUILD STARTED

CI MESSAGE: [41727400]: BUILD STARTED

Greptile Summary

This PR introduces a new thread pool implementation (ThreadPoolBase) with two job types: Job (priority-based) and IncrementalJob (FIFO). The implementation uses POSIX semaphores for worker thread synchronization and atomic counters with condition variables for job completion signaling.

Key implementation details:

• Thread pool uses std::queue for task storage with semaphore-based synchronization
• Jobs wrap tasks in error-catching lambdas that decrement num_pending_tasks_ atomically
• Reentrant design allows submitting sub-jobs from within tasks via WaitOrRunTasks
• TaskBulkAdd batches task additions and defers semaphore release until Submit()

The code is well-structured with comprehensive tests covering basic operations, error handling, reentrancy, and incremental job execution. The synchronization primitives are used correctly throughout.

Confidence Score: 4/5

• This PR is safe to merge with minor considerations for edge cases
• The implementation demonstrates solid concurrent programming practices with proper use of synchronization primitives. Comprehensive tests verify correctness including reentrancy scenarios. Score is 4 rather than 5 due to complexity of concurrent code and potential edge cases in shutdown sequences
• Pay close attention to dali/core/exec/thread_pool_base.cc for the shutdown logic and WaitOrRunTasks reentrant execution flow

Important Files Changed

Sequence Diagram

sequenceDiagram
    participant Client
    participant Job
    participant ThreadPoolBase
    participant Semaphore
    participant WorkerThread
    participant Task

    Note over Client,Task: Normal Job Execution Flow
    Client->>Job: AddTask(runnable, priority)
    Job->>Job: Store in tasks_ multimap (sorted by priority)
    
    Client->>Job: Run(tp, wait=true)
    Job->>ThreadPoolBase: BeginBulkAdd()
    ThreadPoolBase-->>Job: TaskBulkAdd object
    
    loop For each task (by priority)
        Job->>ThreadPoolBase: batch.Add(task)
        ThreadPoolBase->>ThreadPoolBase: AddTaskNoLock() [mutex held]
        ThreadPoolBase->>ThreadPoolBase: tasks_.push()
    end
    
    Job->>ThreadPoolBase: batch.Submit() [destructor]
    ThreadPoolBase->>Semaphore: release(tasks_added)
    
    Semaphore-->>WorkerThread: wake up
    WorkerThread->>Semaphore: acquire()
    WorkerThread->>ThreadPoolBase: PopAndRunTask()
    ThreadPoolBase->>Task: execute()
    Task->>Job: decrement num_pending_tasks_
    
    alt Last task completed
        Task->>Job: DoNotify()
        Job->>Job: num_pending_tasks_.notify_all()
        Job->>Job: cv_.notify_all()
        Job->>Job: running_ = false
    end
    
    Job->>Job: Wait()
    
    alt Called from thread pool
        Job->>ThreadPoolBase: WaitOrRunTasks(cv_, condition)
        Note over ThreadPoolBase: Reentrant: execute tasks while waiting
        loop Until condition met
            ThreadPoolBase->>ThreadPoolBase: PopAndRunTask()
        end
    else Called from external thread
        Job->>Job: num_pending_tasks_.wait()
        Note over Job: Atomic wait on counter
    end
    
    Job-->>Client: Return (rethrow errors if any)

CI MESSAGE: [41726080]: BUILD FAILED

CI MESSAGE: [41730522]: BUILD STARTED